package com.lzq.mq.broker.netty.broker;

import com.alibaba.fastjson2.JSON;
import com.lzq.mq.broker.event.model.*;
import com.lzq.mq.common.coder.TcpMsg;
import com.lzq.mq.common.dto.*;
import com.lzq.mq.common.enums.BrokerEventCode;
import com.lzq.mq.common.event.EventBus;
import com.lzq.mq.common.event.model.Event;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;

@Slf4j
@ChannelHandler.Sharable
public class BrokerServerHandler extends SimpleChannelInboundHandler<TcpMsg> {

    private EventBus eventBus;

    public BrokerServerHandler(EventBus eventBus) {
        this.eventBus = eventBus;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, TcpMsg tcpMsg) throws Exception {
        int code = tcpMsg.getCode();
        byte[] body = tcpMsg.getBody();

        Event event = null;
        if (BrokerEventCode.PUSH_MSG.getCode() == code) {
            MessageDTO messageDTO = JSON.parseObject(body, MessageDTO.class);
            PushMsgEvent pushMsgEvent = new PushMsgEvent();
            pushMsgEvent.setMessageDTO(messageDTO);
            pushMsgEvent.setMsgId(messageDTO.getMsgId());
            //log.info("收到消息推送内容:{},message is {}", messageDTO.getMsgId(), JSON.toJSONString(messageDTO));
            event = pushMsgEvent;
        } else if (BrokerEventCode.CONSUME_MSG.getCode() == code) {
            ConsumeMsgReqDTO consumeMsgReqDTO = JSON.parseObject(body, ConsumeMsgReqDTO.class);
            //log.info("收到消费消息请求:{}", consumeMsgReqDTO.getMsgId());
            InetSocketAddress inetSocketAddress = (InetSocketAddress) channelHandlerContext.channel().remoteAddress();
            consumeMsgReqDTO.setIp(inetSocketAddress.getHostString());
            consumeMsgReqDTO.setPort(inetSocketAddress.getPort());
            ConsumeMsgEvent consumeMsgEvent = new ConsumeMsgEvent();
            consumeMsgEvent.setConsumeMsgReqDTO(consumeMsgReqDTO);
            consumeMsgEvent.setMsgId(consumeMsgReqDTO.getMsgId());
            event = consumeMsgEvent;
        } else if (BrokerEventCode.CONSUME_SUCCESS_MSG.getCode() == code) {
            ConsumeMsgAckReqDTO consumeMsgAckReqDTO = JSON.parseObject(body, ConsumeMsgAckReqDTO.class);
            InetSocketAddress inetSocketAddress = (InetSocketAddress) channelHandlerContext.channel().remoteAddress();
            consumeMsgAckReqDTO.setIp(inetSocketAddress.getHostString());
            consumeMsgAckReqDTO.setPort(inetSocketAddress.getPort());
            ConsumeMsgAckEvent consumeMsgAckEvent = new ConsumeMsgAckEvent();
            consumeMsgAckEvent.setConsumeMsgAckReqDTO(consumeMsgAckReqDTO);
            consumeMsgAckEvent.setMsgId(consumeMsgAckReqDTO.getMsgId());
            event = consumeMsgAckEvent;
        } else if (BrokerEventCode.CREATE_TOPIC.getCode() == code) {
            CreateTopicReqDTO createTopicReqDTO = JSON.parseObject(body,CreateTopicReqDTO.class);
            CreateTopicEvent createTopicEvent= new CreateTopicEvent();
            createTopicEvent.setCreateTopicReqDTO(createTopicReqDTO);
            event = createTopicEvent;
        } else if(BrokerEventCode.START_SYNC_MSG.getCode() == code) {
            StartSyncReqDTO startSyncReqDTO = JSON.parseObject(body, StartSyncReqDTO.class);
            StartSyncEvent startSyncEvent = new StartSyncEvent();
            startSyncEvent.setMsgId(startSyncReqDTO.getMsgId());
            event = startSyncEvent;
        }

        if (event == null) {
            log.error("未知的BrokerService请求:{}", JSON.toJSONString(body));
            return;
        }
        event.setChannelHandlerContext(channelHandlerContext);
        eventBus.publish(event);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        log.info("新的BrokerService连接({})", ctx.channel().remoteAddress());
    }

    /**
     * 断开连接处理
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        super.channelInactive(ctx);

        // 如果依赖任务剔除节点, 会有三个心跳周期的延迟
        // 想要做到链接断开, 立即剔除, 可以通过这个回调函数发送UnRegister时间
        log.info("消费者断开连接:{}", ctx.channel().remoteAddress());
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("连接异常: remoteAddress={}", ctx.channel().remoteAddress(), cause);
        super.exceptionCaught(ctx, cause);
    }
}
